package cn.adelyn.blog.blogmanager.service;

import cn.adelyn.blog.api.file.feign.FileFeignClient;
import cn.adelyn.blog.api.file.feign.PicFeignClient;
import cn.adelyn.blog.api.search.feign.BlogSearchFeignClient;
import cn.adelyn.blog.api.search.pojo.bo.DeleteBlogBO;
import cn.adelyn.blog.api.search.pojo.bo.InsertBlogBO;
import cn.adelyn.blog.blogmanager.dao.BlogDAOService;
import cn.adelyn.blog.blogmanager.dao.BlogFileDAOService;
import cn.adelyn.blog.blogmanager.dao.BlogPicDAOService;
import cn.adelyn.blog.blogmanager.pojo.bo.InsertBlogDecryptBO;
import cn.adelyn.blog.blogmanager.pojo.dto.BlogSearchPageDTO;
import cn.adelyn.blog.blogmanager.pojo.vo.BlogVO;
import cn.adelyn.common.core.cglib.CglibUtil;
import cn.adelyn.common.core.execption.AdelynException;
import cn.adelyn.common.core.pojo.vo.PageVO;
import cn.adelyn.common.core.response.ResponseEnum;
import cn.adelyn.common.database.pojo.dto.PageDTO;
import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.TtlRunnable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

@Service
@Slf4j
public class BlogService {

    @Autowired
    BlogDAOService blogDAOService;

    @Autowired
    BlogPicDAOService blogPicDAOService;

    @Autowired
    BlogFileDAOService blogFileDAOService;

    @Autowired
    FileFeignClient fileFeignClient;

    @Autowired
    PicFeignClient picFeignClient;

    @Autowired
    BlogSearchFeignClient blogSearchFeignClient;

    @Autowired
    @Qualifier("blogManageExecutor")
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    DataSourceTransactionManager transactionManager;

    /**
     * 不同用户线程 key(threadId) 不同，且子线程们 put 操作相同(value 都是 false)，故不需要 ConcurrentHashMap
     * 还是年轻了，hashmap 并发 put 会丢数据，导致没有 parentThreadId 对应的 key, 子线程第二阶段提交前检查时会 nullPoint
     * 那需要 volatile 吗？还是需要的，ConcurrentHashMap put 有锁 get 没锁，子线程 get 还是会存在可见性问题
     * 说到锁 synchronized修改本地内存后，解锁前会将本地内存刷新到主内存中，确保了共享变量的值是最新的，从而保证可见性
     */
    private static volatile Map<Long, Boolean> map = new ConcurrentHashMap<>();
    private static TransmittableThreadLocal<Long> threadContext = new TransmittableThreadLocal<>();

    public void insertBlog(InsertBlogDecryptBO insertBlogDecryptBO) {

        CountDownLatch parentMonitor = new CountDownLatch(1);
        CountDownLatch secondCommitMonitor = new CountDownLatch(2);
        CountDownLatch childMonitor = new CountDownLatch(2);

        Long parentThreadId = Thread.currentThread().getId();

        try {

            // 传给子线程之前就要set，是快照
            threadContext.set(parentThreadId);
            map.put(parentThreadId, true);

            Runnable esRunnable = insertBlogEs(insertBlogDecryptBO, childMonitor, parentMonitor, secondCommitMonitor, threadContext);
            Runnable dbRunnable = insertBlogDb(insertBlogDecryptBO, childMonitor, parentMonitor, secondCommitMonitor, threadContext);

            log.debug("parentThread start: {}, status: {}", threadContext.get(), map);

            threadPoolTaskExecutor.execute(esRunnable);
            threadPoolTaskExecutor.execute(dbRunnable);

            // 等待所有子线程执行完成
            childMonitor.await();
            log.debug("childThreads 1th commit completed, 2th commit start, parentThread: {}, map status: {}", parentThreadId, map);
            // 通知子线程进行第二阶段提交
            parentMonitor.countDown();
            // 等待子线程第二阶段提交结束后进入 finally 清除上下文
            secondCommitMonitor.await();

        } catch (InterruptedException e){
            throw new AdelynException(ResponseEnum.BLOG_INSERT_FAIL, e);
        } finally {
            parentMonitor.countDown();
            map.remove(parentThreadId);
            threadContext.remove();
            log.debug("parentThread run completed: {}, map status: {}", parentThreadId, map);
        }

    }

    private Runnable insertBlogDb(InsertBlogDecryptBO insertBlogDecryptBO, CountDownLatch childMonitor,
                                  CountDownLatch parentMonitor, CountDownLatch secondCommitMonitor,
                                  TransmittableThreadLocal<Long> threadContext) {
        return TtlRunnable.get(() -> {
            try {
                log.debug("db childThread start : {}, parentThread : {}", Thread.currentThread().getId(), threadContext.get());
                TransactionStatus transactionStatus = transactionManager.getTransaction(new DefaultTransactionDefinition());
                // 第一阶段提交
                blogDAOService.insertBlog(insertBlogDecryptBO);

                childMonitor.countDown();

                parentMonitor.await();

                log.debug("db childThread: {}, parentThread : {}, status before 2th commit: {}", Thread.currentThread().getId(), threadContext.get(), map);
                // 第二阶段提交
                if (map.get(threadContext.get())) {
                    log.debug("db childThread: {}, parentThread : {}, commit", Thread.currentThread().getId(), threadContext.get());
                    transactionManager.commit(transactionStatus);
                } else {
                    log.debug("db childThread: {}, parentThread : {}, rollback", Thread.currentThread().getId(), threadContext.get());
                    transactionManager.rollback(transactionStatus);
                }

            } catch (InterruptedException e){
                map.put(threadContext.get(), false);
                throw new AdelynException(ResponseEnum.BLOG_INSERT_FAIL, e.getMessage());
            } finally {
                secondCommitMonitor.countDown();
                childMonitor.countDown();
            }
        });
    }

    private Runnable insertBlogEs(InsertBlogDecryptBO insertBlogDecryptBO, CountDownLatch childMonitor,
                                  CountDownLatch parentMonitor, CountDownLatch secondCommitMonitor,
                                  TransmittableThreadLocal<Long> threadContext) {
        return TtlRunnable.get(() -> {
            try {
                log.debug("es childThread start : {}, parentThread : {}", Thread.currentThread().getId(), threadContext.get());
                InsertBlogBO insertBlogBO = CglibUtil.copy(insertBlogDecryptBO, InsertBlogBO.class);
                // 一阶段提交
                boolean res = blogSearchFeignClient.insertBlog(insertBlogBO).isSuccess();
                if (!res) {
                    log.debug("es childThread insert false : {}, parentThread : {}", Thread.currentThread().getId(), threadContext.get());
                    map.put(threadContext.get(), false);
                }

                childMonitor.countDown();

                parentMonitor.await();

                log.debug("es childThread: {}, parentThread : {}, status before 2th commit: {}", Thread.currentThread().getId(), threadContext.get(), map);
                // 第二阶段提交(没有提交了)
                // es 不支持事务，假装 rollback
                if (!map.get(threadContext.get())){
                    log.debug("es childThread: {}, parentThread : {}, rollback", Thread.currentThread().getId(), threadContext.get());
                    DeleteBlogBO deleteBlogBO = new DeleteBlogBO();
                    deleteBlogBO.setBlogId(insertBlogDecryptBO.getBlogId());
                    blogSearchFeignClient.deleteBlog(deleteBlogBO);
                }

            } catch (InterruptedException e){
                throw new AdelynException(ResponseEnum.BLOG_INSERT_FAIL, e.getMessage());
            } finally {
                secondCommitMonitor.countDown();
                childMonitor.countDown();
            }
        });
    }

    public BlogVO getBlogInfo(Long blogId) {
        return blogDAOService.getBlogInfo(blogId);
    }

    public PageVO<BlogVO> getBlogPage(BlogSearchPageDTO blogSearchPageDTO, PageDTO pageDTO) {
        return blogDAOService.getBlogPage(blogSearchPageDTO, pageDTO);
    }

    public void updateBlog(InsertBlogDecryptBO insertBlogDecryptBO) {
        InsertBlogBO insertBlogBO = CglibUtil.copy(insertBlogDecryptBO, InsertBlogBO.class);
        blogSearchFeignClient.updateBlog(insertBlogBO);
        blogDAOService.updateBlog(insertBlogDecryptBO);
    }

    public void deleteBlog(Long blogId) {

        try {
            List<Long> blogFileList = blogFileDAOService.selectBlogFiles(blogId);
            List<Long> blogPicList = blogPicDAOService.selectBlogPics(blogId);

            if (!blogFileList.isEmpty()) {
                fileFeignClient.deleteFile(blogFileList);
            }
            if (!blogPicList.isEmpty()) {
                picFeignClient.deletePic(blogPicList);
            }

            blogDAOService.deleteBlog(blogId);

            DeleteBlogBO deleteBlogBO = new DeleteBlogBO();
            deleteBlogBO.setBlogId(blogId);
            blogSearchFeignClient.deleteBlog(deleteBlogBO);
        } catch (Exception e) {
            throw new AdelynException(ResponseEnum.BLOG_DELETE_FAIL, e.getMessage());
        }

    }
}
